/*
 * Name     : JobTracker.java
 * Author(s): Jill San Luis, Joshua Kwan
 * Created  : Apr 6, 2011, 12:44:15 AM
 */

import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException.Code;
import org.apache.zookeeper.data.Stat;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.Watcher.Event.EventType;
import java.util.List;
import java.util.ListIterator;

/**
 * The JobTracker class
 *
 * @author  Jill San Luis
 * @author  Joshua Kwan
 */

public class JobTracker {


    /** znode path specific for this class **/
    public static final String ZNODE_PATH = "/jobtracker";

    /** host information for the ZooKeeper */
    private String zkhost;
    
    /** manager for ZooKeeper server connection */
    private ZkConnector zkc;
    
    /** local ZooKeeper object */
    private ZooKeeper zk;
    
    /** Watcher for JobTracker zNode */
    private Watcher jtwatcher;
    
    /** primary or backup status */
    private static final int PRIMARY = 0;
    private static final int BACKUP = 1;  
    private int status = BACKUP;
    
    /**
     * Creates a new instance of JobTracker.
     */
    public JobTracker(String host) {
    	zkc = new ZkConnector();
    	zkhost = host;
        try {
            zkc.connect(host);
        } catch(Exception e) {
            System.out.println("Zookeeper connect "+ e.getMessage());
        }
        
        jtwatcher = new Watcher() {
//        	@Override
        	public void process(WatchedEvent event) {
        		handleEvent(event);
        	}
        };
    }
    
    public static void main(String[] args){
    	if (args.length != 1) {
            System.out.println("Usage: java -classpath lib/zookeeper-3.3.2.jar:lib/log4j-1.2.15.jar:. A zkServer:clientPort");
            return;
        }

        JobTracker jt = new JobTracker(args[0]);
        
        jt.checkpath();
        
        if (jt.status == JobTracker.BACKUP) System.out.println("Backup mode ... waiting");
    	while (jt.status == JobTracker.BACKUP) {
    		Thread.yield();
    	}
    	
    	//now in primary mode
    	jt.primaryMode();
    }

    
    private void checkpath() {
        Stat stat = zkc.exists(ZNODE_PATH, jtwatcher);
        if (stat == null) {              // znode doesn't exist; let's try creating it
            System.out.println("Creating " + ZNODE_PATH);
            Code ret = zkc.create(
            			ZNODE_PATH,         // Path of znode
                        null,           // Data not needed.
                        CreateMode.EPHEMERAL   // Znode type, set to EPHEMERAL.
                        );
            if (ret == Code.OK){
            	System.out.println("JobTracker: Primary mode engaged");
            	status = PRIMARY;
            }
        }
    }

    private void handleEvent(WatchedEvent event) {
        String path = event.getPath();
        EventType type = event.getType();
        if(path.equalsIgnoreCase(ZNODE_PATH)) {
            if (type == EventType.NodeDeleted) {
                System.out.println(ZNODE_PATH + " deleted! Let's go!");       
                checkpath(); // try to become the primary
            }
            if (type == EventType.NodeCreated) {
                System.out.println(ZNODE_PATH + " created!");       
                try{ Thread.sleep(5000); } catch (Exception e) {}
                checkpath(); // re-enable the watch
            }
            if (type == EventType.NodeDataChanged){
            	if (status == PRIMARY){
            		//get data and respond to new requests
            		//Stat stat = new Stat();
            		String jtData = "";
            		try{
            			jtData = zkc.atomicGetAndClearData(ZNODE_PATH);
            			//re-enable the watch
            			checkpath();
                		String[] requests = jtData.split(ZkConnector.DATADELIMITER);
                		for (int i = 0; i < requests.length; i++){
                			String[] request = requests[i].split(" ");
                			if (request.length != 3) System.out.println("JobTracker: bad request - " + requests[i]);
                			else{
                				if (request[1].equals("new")){
                					new JobTrackerNewJobThread(zkc, request[0], request[2]).start();
                				}
                				else if (request[1].equals("query")){
                					new JobTrackerStatusQueryThread(zkc, request[0], request[2]).start();
                				} 
                				else {
                					System.out.println("JobTracker: bad request type - " + request[1]);
                				}
                			}
                		}
            		} catch (IllegalArgumentException e) {
            			System.out.println(e);
            		}
            	}
            }
        }
    }
    
    private void primaryMode(){
        
    	//create PoolCreator
        PoolCreator.createPool(zkc, PoolCreator.JOBPOOL_PATH);
        PoolCreator.createPool(zkc, PoolCreator.WORKERPOOL_PATH);
        
        System.out.println("Persistent pools accounted for");
        
        //check for jobs in process
        recoverCurrentJobs();
        
        //... what else do I need to do here...
        while(true) Thread.yield();
    }
    
    private void recoverCurrentJobs(){
    	zk = zkc.getZooKeeper();
    	try {
    		List<String> currentClients = zk.getChildren(PoolCreator.JOBPOOL_PATH, false);
    		ListIterator cci = currentClients.listIterator();
    		
    		while (cci.hasNext()){
    			String clientID = cci.next().toString();
    			String clientPath = PoolCreator.JOBPOOL_PATH + "/" + clientID;
    			List<String> currentJobs = zk.getChildren(clientPath, false);
    			ListIterator cji = currentJobs.listIterator();
    			while (cji.hasNext()){
    				String jobHash = cji.next().toString();
    				String jobPath = clientPath + "/" + jobHash; 
    				List<String> jobTasks = zk.getChildren(jobPath, false);
    				if (jobTasks.size() > 0) {
    					//recovery is needed!
    					new JobTrackerRecoveryThread(zkc, clientID, jobHash).start();
    				}
    			}
    		}
    	} catch (KeeperException e) {
    		
    	} catch (InterruptedException e) {
    		
    	}
    	
    }
}
